//
// Created by Soar Qin on 2017/8/25.
// Copyright (c) 2017 Soar Qin. All rights reserved.
//

#include "common.hh"

#include "udp_endpoint.hh"

#include "ip_addr.hh"
#include "loop.hh"

#include "private/udp_endpoint_private.hh"
#include "private/io_event_private.hh"
#include "private/loop_private.hh"

#ifndef _WIN32

#if EAGAIN == EWOULDBLOCK
#define _ERR_IS_EAGAIN(e) \
    ((e) == EAGAIN)
#else
#define _ERR_IS_EAGAIN(e) \
    ((e) == EAGAIN || (e) == EWOULDBLOCK)
#endif

/* True iff e is an error that means a read/write operation can be retried. */
#define _ERR_RW_RETRIABLE(e)              \
    ((e) == EINTR || _ERR_IS_EAGAIN(e))
/* True iff e is an error that means an connect can be retried. */
#define _ERR_CONNECT_RETRIABLE(e)         \
    ((e) == EINTR || (e) == EINPROGRESS)
/* True iff e is an error that means a accept can be retried. */
#define _ERR_ACCEPT_RETRIABLE(e)          \
    ((e) == EINTR || _ERR_IS_EAGAIN(e) || (e) == ECONNABORTED)

/* True iff e is an error that means the connection was refused */
#define _ERR_CONNECT_REFUSED(e)                   \
    ((e) == ECONNREFUSED)

#else
/* Win32 */

#define _ERR_IS_EAGAIN(e) \
    ((e) == WSAEWOULDBLOCK || (e) == EAGAIN)

#define _ERR_RW_RETRIABLE(e)                  \
    ((e) == WSAEWOULDBLOCK ||                   \
        (e) == WSAEINTR)

#define _ERR_CONNECT_RETRIABLE(e)                 \
    ((e) == WSAEWOULDBLOCK ||                   \
        (e) == WSAEINTR ||                      \
        (e) == WSAEINPROGRESS ||                    \
        (e) == WSAEINVAL)

#define _ERR_ACCEPT_RETRIABLE(e)          \
    _ERR_RW_RETRIABLE(e)

#define _ERR_CONNECT_REFUSED(e)                   \
    ((e) == WSAECONNREFUSED)

#endif

namespace core::async {

static void udpWriteCB(evutil_socket_t fd, short events, void *arg) {
    (void) fd;
    auto *e = static_cast<IoEvent*>(static_cast<UDPEndpoint *>(arg));
    e->onEvent(events);
}

UDPEndpoint::UDPEndpoint(Loop &loop) : IoEvent(loop), udp_(new UDPEndpointContext) {
}

UDPEndpoint::~UDPEndpoint() {
    if (udp_->evwrite) {
        event_free(udp_->evwrite);
        udp_->evwrite = nullptr;
    }
    for (auto &p: udp_->pendings) {
        free(p.buf);
    }
    udp_->pendings.clear();
    delete udp_;
    udp_ = nullptr;
    if (context_->fd >= 0) {
        evutil_closesocket(context_->fd);
        context_->fd = -1;
    }
}

bool UDPEndpoint::bind(const IPAddr &addr) {
    if (!addr.ok()) return false;
    close();
    isv6_ = addr.isIPv6();
    evutil_socket_t fd = socket(isv6_ ? AF_INET6 : AF_INET, SOCK_DGRAM, 0);
    if (fd < 0) return false;
    int socklen;
    auto *sockaddr = addr.getAddr(socklen);
    if (::bind(fd, sockaddr, socklen) == 0 && startInternal(fd, EV_READ | EV_PERSIST)) {
        evutil_make_socket_nonblocking(fd);
        udp_->evwrite = event_new(loop_.getContext()->evbase, fd, EV_WRITE, udpWriteCB, this);
        return true;
    }
    evutil_closesocket(fd);
    return false;
}

bool UDPEndpoint::bind(const std::string &address) {
    return bind(IPAddr(address));
}

bool UDPEndpoint::bind(const std::string &ip, uint16_t port) {
    return bind(IPAddr(ip, port));
}

int UDPEndpoint::write(const void *buf, size_t size, const IPAddr &to) {
    if (size == 0) return 0;
    bool pending = event_pending(udp_->evwrite, EV_WRITE, nullptr);
    if (pending || !udp_->pendings.empty()) {
        void *data = malloc(size);
        memcpy(data, buf, size);
        udp_->pendings.push_back(UDPPendingPacket {data, size, to});
        if (!pending)
            event_add(udp_->evwrite, nullptr);
        return 0;
    }
    int slen;
    const auto *saddr = to.getAddr(slen);
    int n = sendto(context_->fd, (const char*)buf, size, 0, saddr, slen);
    if (n < 0) {
        auto err = evutil_socket_geterror(context_->fd);
        if (_ERR_RW_RETRIABLE(err)) {
            void *data = malloc(size);
            memcpy(data, buf, size);
            udp_->pendings.push_back(UDPPendingPacket {data, size, to});
            event_add(udp_->evwrite, nullptr);
            return 0;
        }
        close();
        return -1;
    }
    onWritten();
    return n;
}

bool UDPEndpoint::close() {
    if (!IoEvent::close()) return false;
    onClosed();
    return true;
}

void UDPEndpoint::onEvent(short events) {
    if (events & EV_READ) {
        IPAddr from;
        struct sockaddr *saddr;
        ev_socklen_t slen;
        std::tie(saddr, slen) = from.mutableAddr();
        char buf[PACKET_SIZE_MAX];
        int n = recvfrom(context_->fd, buf, PACKET_SIZE_MAX, 0, saddr, &slen);
        if (n == 0) {
            close();
            return;
        }
        if (n < 0) {
            auto err = evutil_socket_geterror(context_->fd);
            if (_ERR_RW_RETRIABLE(err))
                return;
            close();
            return;
        }
        onRead(buf, (size_t)n, from);
    }
    if (events & EV_WRITE) {
        if (udp_->pendings.empty()) return;
        do {
            auto &p = udp_->pendings.front();
            int slen;
            const auto *saddr = p.addr.getAddr(slen);
            if (sendto(context_->fd, (const char*)p.buf, p.size, 0, saddr, slen) >= 0) {
                free(p.buf);
                udp_->pendings.pop_front();
                continue;
            }
            auto err = evutil_socket_geterror(context_->fd);
            if (_ERR_RW_RETRIABLE(err)) {
                event_add(udp_->evwrite, nullptr);
                return;
            }
            close();
            return;
        } while(!udp_->pendings.empty());
        onWritten();
    }
}

}
